package org.yscz.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class MqttCustomClient {

    @Value("${mqtt.url}")
    private String url;

    @Value("${mqtt.sender.clientId}")
    private String clientId;

    @Value("${mqtt.sender.defaultTopic}")
    private String defaultTopic;

    @Value("${mqtt.sender.faceRecognitionResultTopic}")
    private String faceRecognitionResultTopic;

    @Value("${mqtt.sender.imageQualityResultTopic}")
    private String imageQualityResultTopic;

    private MqttClient mqttClient;
    private MqttConfig mqttConfig = new MqttConfig();

    /**
     * true为非持久订阅
     * <p>
     * 方法实现说明 断线重连方法，如果是持久订阅，重连是不需要再次订阅，如果是非持久订阅，重连是需要重新订阅主题 取决于options.setCleanSession(true);
     * <p>
     * 就是这里的clientId，服务器用来区分用户的，不能重复,clientId不能和发布的clientId一样，否则会出现频繁断开连接和重连的问题
     * 不仅不能和发布的clientId一样，而且也不能和其他订阅的clientId一样，如果想要接收之前的离线数据，这就需要将client的 setCleanSession
     * 设置为false，这样服务器才能保留它的session，再次建立连接的时候，它就会继续使用这个session了。 这时此连接clientId 是不能更改的。
     * 但是其实还有一个问题，就是使用热部署的时候还是会出现频繁断开连接和重连的问题，可能是因为刚启动时的连接没断开，然后热部署的时候又进行了重连，重启一	  *   下就可以了
     * System.currentTimeMillis()
     *
     * @throws MqttException
     */
    public void connect() throws MqttException {
        //防止重复创建MQTTClient实例
        if (mqttClient == null) {
            // MemoryPersistence设置clientid的保存形式，默认为以内存保存
            mqttClient = new MqttClient(url, clientId, new MemoryPersistence());
            //如果是订阅者则添加回调类，发布不需要
            mqttClient.setCallback(new PubCallBack(MqttCustomClient.this));
        }
        MqttConnectOptions options = mqttConfig.getOptions();
        //判断拦截状态，这里注意一下，如果没有这个判断，是非常坑的
        if (!mqttClient.isConnected()) {
            mqttClient.connect(options);
        } else {
            mqttClient.disconnect();
            mqttClient.connect(mqttConfig.getOptions(options));
        }
        log.info("连接成功");
    }

    public void init() {
        try {
            connect();
            subscribe(faceRecognitionResultTopic);
            subscribe(imageQualityResultTopic);
            subscribe(defaultTopic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 订阅某个主题，qos默认为0
     *
     * @param topic .
     */
    public void subscribe(String topic) {
        subscribe(topic, 0);
    }

    /**
     * 订阅某个主题
     *
     * @param topic .
     * @param qos   .
     */
    public void subscribe(String topic, int qos) {
        try {
            mqttClient.subscribe(topic, qos);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发布，默认qos为0， 非持久化
     *
     * @param topic
     * @param pushMessage
     * @return
     */
    public boolean publish(String topic, String pushMessage) {
        return publish(0, false, topic, pushMessage);
    }

    /**
     * 发布
     *
     * @param qos
     * @param retained
     * @param topic
     * @param pushMessage
     */
    public boolean publish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        MqttTopic mTopic = mqttClient.getTopic(topic);
        if (null == mTopic) {
            log.info("topic not exist");
            return false;
        }
        MqttDeliveryToken token = null;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttException e) {
            e.printStackTrace();
        }
        return token != null && token.isComplete();
    }
}
